package com.carrot.uaa.infrastructure.kafka;

import cn.hutool.core.util.IdUtil;
import cn.hutool.json.JSONUtil;
import com.carrot.common.core.domain.R;
import com.carrot.uaa.entity.KafkaEvent;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

/**
 * kafka消息生产者
 */
@Service("kafkaProducer")
@Slf4j
@AllArgsConstructor
public class KafkaProducer implements IUaaProduce<KafkaEvent>{

    private final KafkaTemplate<String,String> kafkaTemplate;

    @Override
    public R<?> send(KafkaEvent kafkaEvent) {
        String simpleId = IdUtil.simpleUUID();
        String msg = JSONUtil.toJsonStr(kafkaEvent.getMsg());
        String topic = kafkaEvent.getTopic();

        ProducerRecord<String,String> producerRecord=
                new ProducerRecord<>(topic,simpleId,msg);
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(producerRecord);

        future.thenApplyAsync(result -> {
            log.info(topic+ "-生产者 发送消息成功：" + result.getRecordMetadata());
            return "";
        }).exceptionallyAsync(ex -> {
            log.info(topic+ "-生产者 发送消息失败："+ex.getMessage());
            return "";
        });

        return R.ok();
    }
}
